package cn.linjianhui.demo.flink;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class DataStreamTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Integer> input = env.fromElements(1, 2, 3);
//        input.map(i -> Tuple2.of(i, i))
//                .returns(Types.TUPLE(Types.INT, Types.INT))
//                .print();
        input.flatMap((Integer number, Collector<String> out) -> {
            StringBuilder builder = new StringBuilder();
            for(int i = 0; i < number; i++) {
                builder.append("a");
                out.collect(builder.toString());
            }
        })
        .returns(Types.STRING)
        .print();

        env.execute();
    }
}
